package com.wlyuan.index.domain.index.service.reindex;

import cn.hutool.core.date.StopWatch;
import com.alibaba.fastjson.JSON;
import com.wlyuan.core.domain.valueobj.CombineId;
import com.wlyuan.core.domain.valueobj.TenantId;
import com.wlyuan.core.domain.valueobj.WaybillId;
import com.wlyuan.index.api.dto.waybill.WaybillReindexAllDTO;
import com.wlyuan.index.api.dto.waybill.WaybillReindexDTO;
import com.wlyuan.index.domain.index.factory.WaybillIndexFactory;
import com.wlyuan.index.domain.index.service.WaybillIndexRenewService;
import com.wlyuan.index.domain.source.factory.SourceFactory;
import com.wlyuan.index.domain.source.repository.SourceRepository;
import com.wlyuan.index.domain.source.valueobj.SourceId;
import com.wlyuan.index.remote.facade.BillFacade;
import com.wlyuan.index.remote.facade.QuotationFacade;
import com.wlyuan.index.remote.facade.WaybillFacade;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import lombok.var;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author yuanjie
 */
@Service
@Slf4j
@RequiredArgsConstructor
public class WaybillReindexService {
    private final WaybillIndexFactory indexFactory;
    private final WaybillIndexRenewService renewService;
    private final SourceFactory sourceFactory;
    private final SourceRepository sourceRepository;
    private final WaybillFacade waybillFacade;
    private final QuotationFacade quotationFacade;
    private final BillFacade billFacade;
    private static final ExecutorService EXECUTOR = new ThreadPoolExecutor(4,
            4, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>(),
            new CustomizableThreadFactory("waybill-reindex-pool"));

    public void reindex(WaybillId waybillId) {
        var watch = new StopWatch();
        try {
            // 移除bounds关系
            sourceRepository.unbound(new SourceId(waybillId));

            watch.start("查询运单信息");
            var waybill = waybillFacade.getWaybillById(waybillId);
            if (null == waybill) {
                logger.error("运单不存在: {}", waybillId);
                return;
            }
            watch.stop();

            // 运单信息
            watch.start("保存运单信息");
            sourceFactory.create(waybill);
            watch.stop();

            // 支付信息
            watch.start("查询结算信息");
            var balances = billFacade.listBalances(waybillId);
            watch.stop();
            if (CollectionUtils.isNotEmpty(balances)) {
                watch.start("保存结算信息");
                balances.forEach(sourceFactory::create);
                watch.stop();
            }

            // 运单报价信息
            watch.start("查询报价信息");
            var quotations = quotationFacade.listQuotations(waybillId);
            watch.stop();
            if (CollectionUtils.isNotEmpty(quotations)) {
                watch.start("保存报价信息");
                quotations.forEach(sourceFactory::create);
                watch.stop();
            }

            // 合单报价信息
            if (null != waybill.getCombineId()) {
                var combineId = new CombineId(waybill.getCombineId());
                watch.start("查询报价信息~");
                quotations = quotationFacade.listQuotations(combineId);
                watch.stop();
                if (CollectionUtils.isNotEmpty(quotations)) {
                    watch.start("保存报价信息~");
                    quotations.forEach(sourceFactory::create);
                    watch.stop();
                }
            }

            // 账单信息
            watch.start("查询账单信息");
            var bills = billFacade.listBills(waybillId);
            watch.stop();
            if (CollectionUtils.isNotEmpty(bills)) {
                watch.start("保存账单信息");
                bills.forEach(sourceFactory::create);
                watch.stop();
            }

            // 先删除索引
            watch.start("删除运单索引");
            indexFactory.remove(waybillId);
            watch.stop();
            // 更新索引
            renewService.add(waybillId);
        } catch (Exception e) {
            logger.error("重建运单索引失败: {}", waybillId, e);
        } finally {
            logger.info("重建运单数据耗时: {} {}", waybillId, watch.prettyPrint(TimeUnit.MILLISECONDS));
        }
    }

    public void reindex(WaybillReindexDTO reindex) {
        var reindexJson = JSON.toJSON(reindex);
        logger.info("重建运单索引: {}", reindexJson);
        CompletableFuture.runAsync(() -> {
            var page = 1;
            var size = 1000;
            var totalSize = new AtomicLong(0);
            var waybillIds = waybillFacade.listIds(reindex, page++, size);
            var completedSize = new AtomicLong(0L);
            while (CollectionUtils.isNotEmpty(waybillIds)) {
                logger.info("创建重建运单索引任务: {} {} {}", reindexJson, page - 1, size);
                this.reindexAsync(waybillIds, completedSize);
                totalSize.addAndGet(waybillIds.size());
                waybillIds = waybillFacade.listIds(reindex, page++, size);
            }
            logger.info("创建重建运单索引任务完成: {} {} {}", reindexJson, page - 2, totalSize.get());
        });
    }

    public void reindex(TenantId tenantId) {
        WaybillReindexDTO reindex = new WaybillReindexDTO();
        reindex.setTenantId(tenantId.getId());
        this.reindex(reindex);
    }

    // FIXME: 通过批量获取提高性能
    private void reindexAsync(List<WaybillId> waybillIds, AtomicLong completedSize) {
        CompletableFuture.runAsync(() -> {
            if (CollectionUtils.isEmpty(waybillIds)) {
                return;
            }
            StopWatch watch = new StopWatch();
            watch.start();
            waybillIds.forEach(waybillId -> {
                try {
                    this.reindex(waybillId);
                } catch (Exception e) {
                    logger.error("重建运单索引失败: {}", waybillId, e);
                }
            });
            watch.stop();
            completedSize.addAndGet(waybillIds.size());
            logger.info("重建运单索引任务完成, 耗时: {}ms {} {}", watch.getTotalTimeMillis(),
                    waybillIds.size(), completedSize.get());
        }, EXECUTOR);
    }

    public void reindexAll(List<WaybillId> waybillIds) {
        logger.info("批量重建运单索引数量: {}", waybillIds.size());
        var completedSize = new AtomicLong(0L);
        this.reindexAsync(waybillIds, completedSize);
    }

    public void reindexAll(WaybillReindexAllDTO reindexAll) {
        WaybillReindexDTO reindex = new WaybillReindexDTO();
        BeanUtils.copyProperties(reindexAll, reindex);
        this.reindex(reindex);
    }
}
